package com.ds.lens.data.service.esservice;

import com.google.common.collect.Iterables;
import com.google.gson.Gson;
import com.ds.lens.data.common.asserts.Asserts;
import com.ds.lens.data.common.constant.Constants;
import com.ds.lens.data.common.document.dto.ForecastActualData;
import com.ds.lens.data.common.document.dto.ForecastActualDataPoint;
import com.ds.lens.data.common.document.dynamic.DynamicDetectionResult;
import com.ds.lens.data.kafka.KafkaProducer;
import com.ds.lens.data.vo.out.v2.metric.DynamicDetectionResultOutVO;
import lombok.extern.slf4j.Slf4j;
import org.elasticsearch.index.query.BoolQueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.domain.PageRequest;
import org.springframework.data.domain.Sort;
import org.springframework.data.elasticsearch.core.query.NativeSearchQueryBuilder;
import org.springframework.data.elasticsearch.core.query.SearchQuery;
import org.springframework.stereotype.Service;

import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Optional;

/**
 * Description:
 *
 * @author WeiShaoying
 * @date 2020-05-14
 */
@Slf4j
@Service
public class DynamicDetectionResultService {

    @Autowired
    KafkaProducer kafkaProducer;

    @Value("${forecast.actual.data.kafka.topic}")
    String forecastActualTopic;


    @Autowired
    DynamicDetectionResultOperateService operateService;

    public DynamicDetectionResult markResult(String id) {
        BoolQueryBuilder queryBuilder = QueryBuilders.boolQuery();
        queryBuilder.must(QueryBuilders.termQuery("id", id));

        DynamicDetectionResult detectionResult;
        Optional<DynamicDetectionResult> result = operateService.getIndexRepository().findById(id);
        Asserts.assertTrue(result.isPresent(), Constants.ErrorCode.DATA_NOT_EXIST);
        detectionResult = result.get();
        if (detectionResult.getIsAnomaly() == 1) {
            detectionResult.setIsFalseAlert(true);
        } else {
            detectionResult.setIsAbnormal(true);
        }
        detectionResult = operateService.getIndexRepository().index(detectionResult);
        //send to forecast
        sendToForecast(detectionResult);
        return detectionResult;
    }

    private void sendToForecast(DynamicDetectionResult detectionResult) {
        ForecastActualData input = new ForecastActualData();
        input.setSubTopic(detectionResult.getSubTopic());
        ForecastActualDataPoint actualDataPoint = new ForecastActualDataPoint();
        actualDataPoint.setTs(detectionResult.getTimestamp());
        actualDataPoint.setValue(detectionResult.getActual());
        actualDataPoint.setIsAbnormal(detectionResult.getIsAbnormal());
        actualDataPoint.setIsFalseAlert(detectionResult.getIsFalseAlert());
        input.setData(Collections.singletonList(actualDataPoint));
        kafkaProducer.sendMessageToForecast(forecastActualTopic, null, new Gson().toJson(input));
    }

    public List<DynamicDetectionResultOutVO> queryDynamicDetectionResultOutVOs(Long monitorId, Integer version, Long startTime, Long endTime, String metricBy) {

        BoolQueryBuilder queryBuilder = QueryBuilders.boolQuery();
        queryBuilder.must(QueryBuilders.termQuery("monitorId", monitorId))
                .must(QueryBuilders.termQuery("version", version))
                .must(QueryBuilders.termQuery("metricBy", metricBy))
                .must(QueryBuilders.rangeQuery("timestamp").from(startTime).to(endTime));
        int limit = (int) ((endTime - startTime) / 60000);
        SearchQuery query = new NativeSearchQueryBuilder()
                .withQuery(queryBuilder).withPageable(PageRequest.of(0, limit, Sort.by("timestamp").ascending()))
//                .withSort(SortBuilders.fieldSort("timestamp").order(SortOrder.ASC))
                .build();
        Iterable<DynamicDetectionResult> results = operateService.getIndexRepository().search(query);
        if (Iterables.isEmpty(results)) {
            return Collections.emptyList();
        }
        List<DynamicDetectionResultOutVO> list = new ArrayList<>();
        results.forEach(detectionResult -> {
            DynamicDetectionResultOutVO outVO = new DynamicDetectionResultOutVO();
            BeanUtils.copyProperties(detectionResult, outVO);
            boolean marker = (detectionResult.getIsAbnormal() != null && detectionResult.getIsAbnormal()) || (detectionResult.getIsFalseAlert() != null && detectionResult.getIsFalseAlert());
            outVO.setMarker(marker);
            list.add(outVO);
        });
        return list;
    }
}
